package org.terracotta.toolkit.samples;

import org.terracotta.api.ClusteringToolkit;
import org.terracotta.api.TerracottaClient;
import org.terracotta.collections.ClusteredMap;

import java.io.*;

/**
 * ClusteredMapDemo sample to show usage of Terracotta Toolkit Clustered Map usage
 *
 */
public class ClusteredMapDemo {

    public static final String MAP_NAME = "SHARED_MAP";

    // Handle to the toolkit
    ClusteringToolkit clusterToolkit = null;

    ClusteredMap<Integer, byte[]> clusteredMap = null;

    public void initializeToolKit(String serverAdd) {
        clusterToolkit = new TerracottaClient(serverAdd).getToolkit();
    }

    public void initializeMap() {
        clusteredMap = clusterToolkit.getMap(MAP_NAME);    
    }

    /**
     *
     * Populate the Map with some data
     *
     */
    protected void populateMap(int objectCount) {
        System.out.println("Populating Map...");
        for(int i = 0; i < objectCount; i++) {
            SharedData data = new SharedData(""+i);
            clusteredMap.put(i + 1, serializeObject(data));
            System.out.print(".");
        }
    }

    protected void consumeMapData(int maxObjectId) {
        Range range = new Range(1, maxObjectId);
        while(true) {
            long key = range.getRandomKey();
            byte[] dataBytes = clusteredMap.get((int)key);
            System.out.println("Key = "+key);
            if(dataBytes == null) {
                System.out.println("OOPS!.. somethings wrong dude...");
                return;
            }
            SharedData data = (SharedData)deserializeObject(dataBytes);
            if(data != null) {
                System.out.println("Data = "+data.getData());
            }
            try {
                // Sleep for 5 secs
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static byte[] serializeObject(Object object) {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
		try {
            ObjectOutputStream oos = new ObjectOutputStream(baos);
            oos.writeObject(object);
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
        return baos.toByteArray();
    }

    public static Object deserializeObject(byte[] take) {
		try {
			return new ObjectInputStream(new ByteArrayInputStream(
					take)).readObject();
		} catch (IOException e) {
			throw new IllegalArgumentException("something is wrong");
		} catch (ClassNotFoundException e) {
			throw new IllegalArgumentException("something is wrong");
		}
	}

    public static void main(String[] args) {
        ClusteredMapDemo sample = new ClusteredMapDemo();
        sample.initializeToolKit(args[0]);
        sample.initializeMap();

        // start it as producer if true, else its a consumer
        if("true".equals(args[1]) ) {
            sample.populateMap(100);
        } else {
            sample.consumeMapData(100);
        }
        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            e.printStackTrace(); 
        }
    }

}
